package com.ecommerce.sink;
import com.ecommerce.model.*;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;import java.sql.DriverManager;import java.sql.PreparedStatement;import java.sql.SQLException;import java.util.Properties;
public class ClickHouseSink<T> extends RichSinkFunction<T> {
    private final String tableName;
    private final Class<T> clazz;
    private transient Connection connection;
    private transient PreparedStatement preparedStatement;
    
    public ClickHouseSink(String tableName, Class<T> clazz) {
        this.tableName = tableName;
        this.clazz = clazz;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        
        String url = "jdbc:clickhouse://localhost:8123/ecommerce_analysis";
        Properties properties = new Properties();
        properties.setProperty("user", "default");
        properties.setProperty("password", "");
        properties.setProperty("socket_timeout", "300000");
        
        connection = DriverManager.getConnection(url, properties);
        
        String sql = getInsertSQL(tableName);
        preparedStatement = connection.prepareStatement(sql);
    }
    
    @Override
    public void invoke(T value, Context context) throws Exception {
        setPreparedStatementParameters(value);
        preparedStatement.addBatch();
        
        // 批量提交
        if (context.currentProcessingTime() % 5000 == 0) {
            preparedStatement.executeBatch();
        }
    }
    
    @Override
    public void close() throws Exception {
        if (preparedStatement != null) {
            preparedStatement.executeBatch();
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
        super.close();
    }
    
    private String getInsertSQL(String tableName) {
        switch (tableName) {
            case "product_ranking":
                return "INSERT INTO product_ranking (product_id, product_name, keywords, shop_name, " +
                       "pay_buyer_count, visitor_count, payment_amount, growth_rate, price, category, " +
                       "shop_type, timestamp, rank, rank_type) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
            case "shop_ranking":
                return "INSERT INTO shop_ranking (shop_name, on_sale_product_count, payment_amount, " +
                       "growth_rate, avg_unit_price, shop_type, category, timestamp, rank, rank_type) " +
                       "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
            case "search_ranking":
                return "INSERT INTO search_ranking (keyword, word_type, search_count, growth_rate, " +
                       "supply_demand_ratio, category, timestamp, rank, rank_type) " +
                       "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
            case "content_ranking":
                return "INSERT INTO content_ranking (content_id, shop_name, content_type, content_sub_type, " +
                       "transaction_amount, view_count, click_count, category, timestamp, rank) " +
                       "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
            case "category_analysis":
                return "INSERT INTO category_analysis (category, transaction_amount, transaction_count, " +
                       "supply_demand_ratio, growth_rate, market_share, timestamp, analysis_type, opportunity_type) " +
                       "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
            default:
                throw new IllegalArgumentException("Unknown table: " + tableName);
        }
    }
    
    private void setPreparedStatementParameters(T value) throws SQLException {
        if (value instanceof ProductData) {
            setProductParameters((ProductData) value);
        } else if (value instanceof ShopData) {
            setShopParameters((ShopData) value);
        } else if (value instanceof SearchWordData) {
            setSearchWordParameters((SearchWordData) value);
        } else if (value instanceof ContentData) {
            setContentParameters((ContentData) value);
        } else if (value instanceof CategoryData) {
            setCategoryParameters((CategoryData) value);
        }
    }
    
    private void setProductParameters(ProductData product) throws SQLException {
        preparedStatement.setLong(1, product.getProductId());
        preparedStatement.setString(2, product.getProductName());
        preparedStatement.setString(3, product.getKeywords());
        preparedStatement.setString(4, product.getShopName());
        preparedStatement.setInt(5, product.getPayBuyerCount());
        preparedStatement.setInt(6, product.getVisitorCount());
        preparedStatement.setDouble(7, product.getPaymentAmount());
        preparedStatement.setDouble(8, product.getGrowthRate());
        preparedStatement.setDouble(9, product.getPrice());
        preparedStatement.setString(10, product.getCategory());
        preparedStatement.setString(11, product.getShopType());
        preparedStatement.setLong(12, product.getTimestamp());
        preparedStatement.setInt(13, product.getRank());
        preparedStatement.setString(14, product.getRankType());
    }
    
    private void setShopParameters(ShopData shop) throws SQLException {
        preparedStatement.setString(1, shop.getShopName());
        preparedStatement.setInt(2, shop.getOnSaleProductCount());
        preparedStatement.setDouble(3, shop.getPaymentAmount());
        preparedStatement.setDouble(4, shop.getGrowthRate());
        preparedStatement.setDouble(5, shop.getAvgUnitPrice());
        preparedStatement.setString(6, shop.getShopType());
        preparedStatement.setString(7, shop.getCategory());
        preparedStatement.setLong(8, shop.getTimestamp());
        preparedStatement.setInt(9, shop.getRank());
        preparedStatement.setString(10, shop.getRankType());
    }
    
    private void setSearchWordParameters(SearchWordData word) throws SQLException {
        preparedStatement.setString(1, word.getKeyword());
        preparedStatement.setString(2, word.getWordType());
        preparedStatement.setInt(3, word.getSearchCount());
        preparedStatement.setDouble(4, word.getGrowthRate());
        preparedStatement.setDouble(5, word.getSupplyDemandRatio());
        preparedStatement.setString(6, word.getCategory());
        preparedStatement.setLong(7, word.getTimestamp());
        preparedStatement.setInt(8, word.getRank());
        preparedStatement.setString(9, word.getRankType());
    }
    
    private void setContentParameters(ContentData content) throws SQLException {
        preparedStatement.setString(1, content.getContentId());
        preparedStatement.setString(2, content.getShopName());
        preparedStatement.setString(3, content.getContentType());
        preparedStatement.setString(4, content.getContentSubType());
        preparedStatement.setDouble(5, content.getTransactionAmount());
        preparedStatement.setInt(6, content.getViewCount());
        preparedStatement.setInt(7, content.getClickCount());
        preparedStatement.setString(8, content.getCategory());
        preparedStatement.setLong(9, content.getTimestamp());
        preparedStatement.setInt(10, content.getRank());
    }
    
    private void setCategoryParameters(CategoryData category) throws SQLException {
        preparedStatement.setString(1, category.getCategory());
        preparedStatement.setDouble(2, category.getTransactionAmount());
        preparedStatement.setInt(3, category.getTransactionCount());
        preparedStatement.setDouble(4, category.getSupplyDemandRatio());
        preparedStatement.setDouble(5, category.getGrowthRate());
        preparedStatement.setDouble(6, category.getMarketShare());
        preparedStatement.setLong(7, category.getTimestamp());
        preparedStatement.setString(8, category.getAnalysisType());
        preparedStatement.setString(9, category.getOpportunityType());
    }}